//
// EpollDispatcher.cpp
//
// $Id: //poco/1.4/Net/src/EpollDispatcher.cpp#1 $
//
// Library: Net
// Package: EpollServer
// Module:  EpollDispatcher
//
// Copyright (c) 2005-2007, Applied Informatics Software Engineering GmbH.
// and Contributors.
//
// SPDX-License-Identifier:	BSL-1.0
//


#include "Poco/Net/EpollDispatcher.h"
#include "Poco/ErrorHandler.h"
#include <memory>
#include "Poco/Net/EpollServer.h"
#include <time.h>
#include <iostream>
using Poco::FastMutex;

namespace Poco {
namespace Net {


EpollDispatcher::EpollDispatcher(Poco::ThreadPool& threadPool,EpollClientFactory& factory,unsigned int timeoutSecond)
:m_threadPool(threadPool),
m_factory(factory),
m_stopped(false),
m_timeout(timeoutSecond),
m_maxFd(POCO_INVALID_SOCKET),
m_queuesize(0)
{
	m_epollServer = NULL;
	m_thread.start(*this);
}


EpollDispatcher::~EpollDispatcher()
{
	stop();
	for(auto socket:m_sockets)
	{
		delete socket.second;
	}
	m_sockets.clear();
}

void EpollDispatcher::enqueue(int fd)
{
	ScopedLock<Poco::FastMutex> lock(m_queueMutex);
	if (fd > 0)
	{
		/// 水平触发和边沿触发共用一个队列，但要处理水平触发时，相同fd过来的消息
		if (m_queue.size() < fd)
		{
			m_queue.resize(fd+10000,POCO_INVALID_SOCKET);
		}
		m_queue[fd] = fd;
		if (m_maxFd < fd)
			m_maxFd = fd;
		++m_queuesize;
	}
}
	
void EpollDispatcher::dequeue(int fd)
{
	ScopedLock<Poco::FastMutex> lock(m_queueMutex);
	if (fd > 0 && m_queue[fd] != POCO_INVALID_SOCKET)
	{
		m_queue[fd]=POCO_INVALID_SOCKET;
		//m_queuesize > 0? --m_queuesize:0;
		--m_queuesize;
		if (fd == m_maxFd)
		{
			--m_maxFd;
		}
		if (m_queue.size() - m_maxFd > 11000)
		{
			m_queue.resize(m_queue.size() - 10000);
		}
	}
}

void EpollDispatcher::stop()
{
	if (!m_stopped)
	{
		m_stopped = true;
		m_thread.join();
	}
}

void EpollDispatcher::setEpollServer(EpollServer* server)
{
	m_epollServer = server;
}

void EpollDispatcher::checkCloseFd(int fd)
{
	if (!m_epollServer)
		return;
	auto itfd = m_timeoutFds.find(fd);
	if (m_timeout > 0)
	{
		if (itfd != m_timeoutFds.end())
		{
			if (time(NULL) - itfd->second >= m_timeout )
			{
				m_epollServer->addTimeoutFd(fd);
				m_timeoutFds.erase(itfd);
			}
		}
		else
		{
			m_timeoutFds[fd] = time(NULL);
		}
	}
	else
	{
		m_epollServer->addTimeoutFd(fd);
		m_timeoutFds.erase(itfd);
	}
}


void EpollDispatcher::run()
{
	while(!m_stopped)
	{
		bool bError = false;
		int fd = POCO_INVALID_SOCKET, i = 0;
		while( i <= m_maxFd && !bError)			// 不加锁取大小，应该没问题
		{
			fd = m_queue[i++];		/// m_queue size已由m_maxFd限制
			if (fd == POCO_INVALID_SOCKET)	continue;	// 因为没加锁
			bool bTimeout = true;
			try
			{
				if (m_sockets.find(fd) == m_sockets.end())
				{
					/// 主要是用一个成员保存EpollClientSocket的生成期，即保存一个Runnable对象的生成期
					/// 如果fd是重新连接后生成的fd，可需要处理的业务不一样，但应该不影响本处理，即不用先delete原来的指针，再新建一个。
					m_sockets[fd] = m_factory.createClientSocket(fd);
				}
				m_threadPool.start(*m_sockets[fd]);
				if (m_timeout > 0 && m_epollServer)
				{
					m_timeoutFds.erase(fd);
				}
				bTimeout = false;
				dequeue(fd);
			}
			catch(NoThreadAvailableException& e)
			{
				bError = true;
				/// 线程不够，不计超时时间
				break;
			}
			catch(Exception& e)
			{
				checkCloseFd(fd);
				continue;
			}
			if (bTimeout)
			{
				checkCloseFd(fd);
			}
		}
		m_threadPool.joinTimeoutWorkingThread(m_timeout);
		Poco::Thread::sleep(1);
	}
}

}
}









